package com.atguigu.day9


import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.api.scala.typeutils.Types
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.co.CoProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector
object RealTimeCheckBill {

  //支付订单事件
  case class OrderEvent(
                         orderId:String,
                         eventType:String,
                         eventTime:Long
                       )

  //第三方支付事件，    weixin   zhifuao
  case class PayEvent(
                       orderId:String,
                       eventType:String,
                       eventTime:Long
                     )
  //未匹配到支付订单事件
  val unmatchedOrders = new OutputTag[String]("unmatchedOrders")

  //未匹配到第三方支付事件
  val unmatchedpays = new OutputTag[String]("unmatchedpays")

  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val orders = env.fromElements(
      OrderEvent("order_1", "pay", 2000L),
      OrderEvent("order_2", "pay", 3000L),
      OrderEvent("order_3", "pay", 4000L)
    ).assignAscendingTimestamps(_.eventTime)
      .keyBy(_.orderId)

    val pays = env.fromElements(
      PayEvent("order_1", "zhifubao", 5000L),
      PayEvent("order_4", "zhifubao", 6000L),
      PayEvent("order_5", "zhifubao", 7000L)
    ).assignAscendingTimestamps(_.eventTime)
      .keyBy(_.orderId)

    val processd = orders
      .connect(pays)
      .process(new MatchFunction)
    processd.print()    //打印对账成功的订单
    processd.getSideOutput(unmatchedOrders).print()   //打印订单支付事件到达，第三方支付事件未到达
    processd.getSideOutput(unmatchedpays).print()     //打印第三方支付事件到达，订单支付事件未到达


    env.execute()


  }

  class MatchFunction extends CoProcessFunction[OrderEvent,PayEvent,String]{

    lazy val orderState = getRuntimeContext.getState(
      new ValueStateDescriptor[OrderEvent]("order-state",Types.of[OrderEvent])
    )
    lazy val payState = getRuntimeContext.getState(
      new ValueStateDescriptor[PayEvent]("pay-state",Types.of[PayEvent])
    )

    //用来处理来自订单支付事件流的元素
    override def processElement1(order: OrderEvent, ctx: CoProcessFunction[OrderEvent, PayEvent, String]#Context, out: Collector[String]): Unit = {
      val pay = payState.value()
      if (pay != null){
        //同样订单ID的第三方支付事件先到了，因为pay不为空
        //而此时订单支付事件也到了，所以对账成功
        //清空payState
        payState.clear()
        out.collect("订单ID为："+order.orderId+"的订单对账成功")
      }else{
        //同样订单ID的第三方支付事件先到，因为pay为null
        //所以先将订单支付事件存下来，等待对账，等待5s
        orderState.update(order)
        ctx.timerService().registerEventTimeTimer(order.eventTime+5000L)
      }

    }

    //处理来自第三方支付事件流的元素
    override def processElement2(pay: PayEvent, ctx: CoProcessFunction[OrderEvent, PayEvent, String]#Context, out: Collector[String]): Unit = {
      val order = orderState.value()

      if(order != null){
        orderState.clear()
        out.collect("订单ID为："+order.orderId+"的订单对账成功")
      }else{
        payState.update(pay)
        ctx.timerService().registerEventTimeTimer(pay.eventTime+5000L)

      }

    }

    override def onTimer(timestamp: Long, ctx: CoProcessFunction[OrderEvent, PayEvent, String]#OnTimerContext, out: Collector[String]): Unit = {
      if (payState.value() != null){
        ctx.output(unmatchedpays,"订单ID是："+payState.value().orderId+"对账失败，订单支付事件没来")
        payState.clear()
      }
      if (orderState.value() !=null){
        ctx.output(unmatchedOrders,"订单ID是："+orderState.value().orderId+"对账失败，第三方支付事件没来")
        orderState.clear()
      }
    }
  }
}
